package net.neptech.service.consumer;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import lombok.extern.slf4j.Slf4j;
import net.neptech.entity.SpeechText;
import net.neptech.entity.vo.dto.KSignalMessage;
import net.neptech.service.ai.AiSummaryService;
import net.neptech.websocket.SignalingWebSocket;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

    /**
     * Kafka 信令消息消费者
     * 负责从 Kafka 接收消息，并调用 WebSocket 服务发送给目标用户
     * ✅ 支持：实时字幕 + AI 会议纪要生成
     */
    @Slf4j
    @Service
    public class SignalingConsumer {

        @Autowired
        private AiSummaryService aiSummaryService;

        @Autowired
        private RedisTemplate<String, Object> redisTemplate;

        //  存储每个会议的语音转写记录，先暂存，后面我在弄到redis里面，不知道整体项目用啥存
        private static final ConcurrentHashMap<String, List<SpeechText>> meetingTranscripts = new ConcurrentHashMap<>();
        /**
         *  初始化会议记录容器
         */
        private void initTranscriptsIfAbsent(String conferenceId) {
            meetingTranscripts.putIfAbsent(conferenceId, new CopyOnWriteArrayList<>());
        }

        /**
         * 监听 Kafka Topic，处理信令消息
         */
        @KafkaListener(topics = "lxy-signaling-top", groupId = "lxy-group")
        public void consume(String message, Acknowledgment ack) {
            try {
                KSignalMessage kMsg = JSON.parseObject(message, KSignalMessage.class);
                if (kMsg == null) return;

                String type = kMsg.getType();
                String conferenceId = kMsg.getConferenceId();

                if ("speech-text".equals(type)) {
                    handleSpeechText(kMsg);
                } else if ("meeting-ended".equals(type)) {
                    handleMeetingEnded(kMsg);
                } else if ("ai-meeting-summary".equals(type)) {
                    //直接广播给所有人
                    JSONObject msg = new JSONObject();
                    msg.put("type", "ai-meeting-summary");
                    msg.put("data", String.valueOf(kMsg.getData()));

                    SignalingWebSocket.broadcast(conferenceId, null, msg.toJSONString());
                    log.info("📢 广播 AI 会议纪要成功 (conf={}): {}字",
                            conferenceId, String.valueOf(kMsg.getData()).length());
                } else {
                    SignalingWebSocket.forwardMessageFromKafka(kMsg);
                }
            } finally {
                ack.acknowledge();
            }
        }

        /**
         * 处理语音转写消息（实时字幕）
         */
        private void handleSpeechText(KSignalMessage kMsg) {

            String conferenceId = kMsg.getConferenceId();
            String fromUserId = kMsg.getFromUserId();
            Object dataObj = kMsg.getData();
            SpeechText st = (SpeechText) kMsg.getData();

            // 存到 Redis list
            String key = "meeting:transcripts:" + conferenceId;
            redisTemplate.opsForList().rightPush(key, st);


            if (!(dataObj instanceof JSONObject data)) {
                log.warn("speech-text 的 data 不是 JSON: {}", dataObj);
                return;
            }

            String text = data.getString("text");
            String doctorName = data.getString("doctorName");
            String role = data.getString("role");

            if (text == null || text.isEmpty()) {
                log.warn("speech-text 缺少 text 字段: {}", data);
                return;
            }

            // 1. 保存到会议记录中（用于 AI 生成）
            initTranscriptsIfAbsent(conferenceId);
            meetingTranscripts.get(conferenceId).add(
                    new SpeechText(fromUserId, doctorName, role, text, System.currentTimeMillis())
            );

            // 2. 广播给房间所有人（实现“实时字幕”）
            JSONObject forwardMsg = new JSONObject();
            forwardMsg.put("type", "speech-text");
            forwardMsg.put("from", fromUserId);
            forwardMsg.put("data", data);

            SignalingWebSocket.broadcast(conferenceId, null, forwardMsg.toJSONString());
            log.info("📢 字幕广播成功: {}... (会议: {})", text.substring(0, Math.min(20, text.length())), conferenceId);
        }

        /**
         * 处理会议结束消息（触发 AI 生成会议纪要）
         */
        private void handleMeetingEnded(KSignalMessage kMsg) {
            String conferenceId = kMsg.getConferenceId();

            // 从 Redis 取逐字稿 + 文件
            List<SpeechText> transcripts = loadTranscriptsFromRedis(conferenceId);
            List<String> files = loadFilesFromRedis(conferenceId);

            // 调用 AI
            String summary = aiSummaryService.generateMeetingSummary(conferenceId, transcripts, files);

            // 推送到前端
            broadcastAISummary(conferenceId, summary);

        }

        /**
         * 广播 AI 会议纪要给所有人
         */
        private void broadcastAISummary(String conferenceId, String summary) {
            JSONObject msg = new JSONObject();
            msg.put("type", "ai-meeting-summary");
            msg.put("data", summary);
            SignalingWebSocket.broadcast(conferenceId, null, msg.toJSONString());
            log.info("✅ AI 会议纪要已广播 (会议: {})", conferenceId);
        }
        /** 从 Redis 取逐字稿 */
        private List<SpeechText> loadTranscriptsFromRedis(String conferenceId) {
            String key = "meeting:transcripts:" + conferenceId;
            List<Object> objs = redisTemplate.opsForList().range(key, 0, -1);
            if (objs == null) return List.of();
            return objs.stream().map(o -> (SpeechText) o).toList();
        }

        /** 从 Redis 取文件内容 */
        private List<String> loadFilesFromRedis(String conferenceId) {
            String key = "meeting:files:" + conferenceId;
            List<Object> objs = redisTemplate.opsForList().range(key, 0, -1);
            if (objs == null) return List.of();
            return objs.stream().map(Object::toString).toList();
        }
}



